package com.iot.service.mqtt.service.retain;

import com.iot.service.mqtt.model.request.TlMqttPublishReq;
import com.iot.service.mqtt.service.StoreStrategy;
import com.iot.service.mqtt.service.topic.ITopicService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @Author: hszhou
 * @Date: 2025/1/8 14:17
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
public class LocalRetainServiceImpl implements IRetainService, ITopicService {

    private static final Logger logger = LoggerFactory.getLogger(LocalRetainServiceImpl.class);


    public static final ConcurrentHashMap<String, TlMqttPublishReq> RETAIN_MAP = new ConcurrentHashMap<>();

    @Override
    public Mono<Boolean> save(String topic, TlMqttPublishReq req) {
        return Mono.fromSupplier(() -> { RETAIN_MAP.put(topic, req);
            return true;
        });
    }

    @Override
    public Mono<TlMqttPublishReq> find(String topic) {

        return Mono.fromSupplier(()-> RETAIN_MAP.entrySet().stream()
                 .filter(entry -> match(entry.getKey(), topic))
                 .map(Map.Entry::getValue).findFirst()
                .orElse(null));

    }

    @Override
    public Mono<Boolean> clear(String topic) {
       return Mono.fromSupplier(()-> RETAIN_MAP.remove(topic)==null);
    }
}
